package com.o2o.cleaning.month.platform.ebusiness_plat.wangyiyanxuan

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}

import scala.collection.mutable.ArrayBuffer

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   20191/4/18
  * @ Param:  ${PARAM}
  * @ Description:
  *
  * 1.销量：日累计销量
  * 2.分类：swbfirstId,swbfirstName,firstCategoryId,secondCategoryId,thirdCategoryId,fourthCategoryId
  * 3.地址：网易公司的具体地址
  *
  */
object WYYXRelation {


  /** 销量计算
    *
    * @param spark
    * @param rDD
    * @param timeStamp
    * @return
    */
  def caculate(spark: SparkSession, rDD: RDD[String], timeStamp: Long, startTime: Long = 0, endTime: Long = 4133951999L): DataFrame = {

    val rdd = rDD.map(line => {
      val nObject: JSONObject = JSON.parseObject(line)
      val add_to_field = nObject.getJSONArray("add_to_field")

      if (null != add_to_field) {
        var priceText_Loc = 0D
        var sellCount_Loc = 0L

        import scala.collection.mutable.Map
        val sell_map: Map[Long, Long] = Map()

        val price_temp = ArrayBuffer[Double](0)
        val sell_temp = ArrayBuffer[Long](0)


        for (i <- 0 to add_to_field.size() - 1) {
          val add_to_field_i: JSONObject = JSON.parseObject(add_to_field.get(i).toString)

          val price_add = add_to_field_i.getOrDefault("priceText", "-1").toString.toDouble
          val sell_add = add_to_field_i.getOrDefault("sellCount", "-1").toString.toLong
          val crawl_date = add_to_field_i.getOrDefault("crawl_date", "-1").toString.toLong

          if (startTime <= crawl_date && crawl_date <= endTime) {
            if (price_add > 0) price_temp += price_add
            if (i == 0) sell_temp += sell_add
            if (sell_add > 0) sell_temp += sell_add
            sell_map.put(crawl_date, sell_add) // 用于乱序时排序用
          }
        }

        val sell_tuple: Seq[(Long, Long)] = sell_map.toSeq.sortBy(_._1)
        if (price_temp.length > 1) priceText_Loc = price_temp.last.formatted("%.2f").toDouble
        if (sell_temp.length > 1) sellCount_Loc = sell_tuple.last._2 - sell_tuple.head._2
        val sellCount_m = if (sell_temp.length > 1) sell_temp.last - sell_temp(1) else 0

        //***************************修改部分***************************//

//        val sells: Int = (sellCount_Loc * 30 / 27).toInt

        //***************************修改部分***************************//


        nObject.put("priceText", priceText_Loc)
//        nObject.put("sellCount", sells)
        nObject.put("sellCount", sellCount_Loc)
//        nObject.put("sellCount_m", sells)
        nObject.put("sellCount_m", sellCount_m)
//        nObject.put("salesAmount", (priceText_Loc * sells).formatted("%.2f").toDouble)
        nObject.put("salesAmount", (priceText_Loc * sellCount_Loc).formatted("%.2f").toDouble)

      }

      nObject.toString
    })

    spark.read.json(rdd)
  }


  /** 添加地址  并添加平台必需字段
    *
    * @param spark
    * @param dataFrame
    * @param timeStamp
    * @return
    */
  def parseAddress(spark: SparkSession, dataFrame: DataFrame, timeStamp: Long): DataFrame = {

    val rdd = dataFrame.toJSON.rdd.map(line => {
      val nObject: JSONObject = JSON.parseObject(line)

      nObject.put("good_id", nObject.get("good_id").toString)
      nObject.put("platformName", "网易严选")
      nObject.put("platformId", "55")
      nObject.put("platformName_spelling", "wangyiyanxuan")
      nObject.put("brandValueId", "wy100001")
      nObject.put("brandName", "网易严选")
      nObject.put("brandName_cn", "网易严选")
      nObject.put("brandName_en", "网易严选")
      nObject.put("timeStamp", s"${timeStamp}")
      nObject.put("brand_type", "国产品牌")
      //nObject.put("good_id", nObject.get("good_id").toString)
      nObject.put("Base_Info", nObject.getOrDefault("Base_Info", "{}").toString)
      nObject.put("goodRatePercentage", nObject.getOrDefault("goodRatePercentage", "-1").toString.replace("%", ""))

      nObject.put("administrative_region", "华东地区")
      nObject.put("city", "杭州市")
      nObject.put("city_grade", "2")
      nObject.put("city_origin", "杭州市")
      nObject.put("district", "滨江区")
      nObject.put("district_origin", "滨江区")
      nObject.put("economic_division", "2")
      nObject.put("if_city", "1")
      nObject.put("if_district", "2")
      nObject.put("if_state_level_new_areas", "0")
      nObject.put("poor_counties", "0")
      nObject.put("province", "浙江省")
      nObject.put("regional_ID", "330108")
      nObject.put("rural_demonstration_counties", "0")
      nObject.put("rural_ecommerce", "0")
      nObject.put("the_belt_and_road_city", "0")
      nObject.put("the_belt_and_road_province", "2")
      nObject.put("the_yangtze_river_economic_zone_city", "1")
      nObject.put("the_yangtze_river_economic_zone_province", "1")
      nObject.put("urban_agglomerations", "1")
      nObject.put("registration_institution", "杭州市高新区（滨江）市场监督管理局")
      nObject.put("address", "浙江省杭州市滨江区长河街道网商路599号4幢410室")
      nObject.put("name", "杭州网易严选贸易有限公司")

      nObject.toString
    })

    spark.read.json(rdd)
  }


  /**
    * 处理分类
    *
    * 先以categoryId关联后拆分，再关联
    *
    * @param spark
    * @param dataFrame
    * @param cateDF
    * @return
    */
  def parseCate(spark: SparkSession, dataFrame: DataFrame, cateDF: DataFrame): DataFrame = {

    val cate = cateDF.selectExpr("categoryId", "swbfirstId", "swbfirstName", "fourthCategoryId").dropDuplicates("categoryId")
    //1.以categoryId关联
    val data_fourCate = dataFrame.selectExpr("good_id","subCategoryId","title").withColumnRenamed("subCategoryId","categoryId")
      .filter("categoryId is not null").join(cate, Seq("categoryId"), "left")

    //2.拆分fourthCategoryId
    data_fourCate.createOrReplaceTempView("fourCate_v")
    val result_cate = spark.sql(
      """
        |select
        |good_id, swbfirstId, swbfirstName
        |,substr(fourthCategoryId,1,5) as firstCategoryId
        |,substr(fourthCategoryId,1,7) as secondCategoryId
        |,substr(fourthCategoryId,1,9) as thirdCategoryId
        |,fourthCategoryId
        |from
        |(
        |select good_id, swbfirstId, swbfirstName
        |-- 顺序不可改变
        |,case
        |when categoryId ='109256016' then
        |     (case
        |           when title rlike '挂架' then '10022069999'
        |           when title rlike '洗衣机' then '10014010299'
        |           when title rlike '(香炉)|(线香)|(蜡烛)' then '10022159999'
        |           else '10022159999' end
        |     )
        |when categoryId ='1010004' then
        |     (case
        |           when title rlike '(腕表)|(手表)' then '10017030199'
        |           when title rlike '(皮带)|(腰带)' then '10011999999'
        |           else '10011999999' end
        |     )
        |when categoryId ='109256009' then
        |     (case
        |           when title rlike '(清洁)|(洁面)' then '10019010199'
        |           when title rlike '(卸妆)|(湿巾)' then '10019020199'
        |           else '10019019999' end
        |     )
        |when categoryId ='1008008' then
        |     (case
        |           when title rlike '被子' then '10022050499'
        |           when title rlike '蚊帐' then '10022050599'
        |           when title rlike '(毛巾)|(浴巾)' then '10022050299'
        |           when title rlike '枕巾枕芯' then '10022050799'
        |           when title rlike '(床垫)|(床褥)|(感垫)' then '10022059999'
        |           when title rlike '(凉席)' then '10022050699'
        |           else '10022059999' end
        |     )
        |when categoryId ='109206009' then
        |     (case
        |           when title rlike '(鸡蛋)|(鸭蛋)' then '10021030499'
        |           when title rlike '(鱼)|(虾)|(鳕)' then '10021030399'
        |           when title rlike '(番茄)' then '10021030299'
        |           when title rlike '(肉)|(鸡)|(牛)|(火腿)' then '10021030599'
        |           when title rlike '(水果)|(果)|(瓜)|(柚)|(桃)|(枣)|(柠檬)|(栗)|(薯)|(椰)|(樱桃)|(杏)' then '10021030199'
        |           else '10021039999' end
        |     )
        |when categoryId ='109206007' then
        |     (case
        |           when title rlike '(油)' then '10021020199'
        |           when title rlike '(绿豆)|(杂粮)|(小豆)|(五谷米)|(薏仁米)|(黑米)' then '10021020599'
        |           when title rlike '(大米)|(禾糯)|(贡米)' then '10021020399'
        |           when title rlike '(拉面)|(拌面)|(酱面)' then '10021020199'
        |           when title rlike '(方便面)' then '10021020799'
        |           when title rlike '(面)' then '10021020499'
        |           else '10021029999' end
        |     )
        |when categoryId ='1005011' then
        |     (case
        |           when title rlike '(巧克力)' then '10021010199'
        |           when title rlike '(坚果)|(板栗)' then '10021010299'
        |           when title rlike '(糖)' then '10021010399'
        |           when title rlike '(肉)' then '10021010699'
        |           when title rlike '(薯条)|(薯片)|(蟹片)|(锅巴)|(爆米花)' then '10021010899'
        |           --when title rlike '' then ''
        |           else '10021019999' end
        |     )
        |when categoryId ='1008006' then
        |     (case
        |           when title rlike '(音像)' then '10023049999'
        |           when title rlike '(投影仪)' then '10016040199'
        |           when title rlike '(音像)|(音箱)|(音响)' then '10015059999'
        |           when title rlike '耳机' then '10015019999'
        |           when title rlike '相机' then '10015039999'
        |           else '10023999999' end
        |     )
        |when categoryId ='109254052' then
        |     (case
        |           when title rlike '瑜伽' then '10013090299'
        |           else '10013099999' end
        |     )
        |else fourthCategoryId end fourthCategoryId
        |from fourCate_v
        |)
      """.stripMargin)

    //3.关联分类
    val dataCate: DataFrame = dataFrame.drop("swbfirstId", "swbfirstName","firstCategoryId", "secondCategoryId", "thirdCategoryId", "fourthCategoryId")
      .join(result_cate, Seq("good_id"), "left").dropDuplicates("good_id")

    dataCate
  }


}
